#include "groupByRequestId.h"

namespace sp::Tool::Group{

void registe_groupByRequestId(){
    groupByRequestId::registe();
}

groupByRequestId::groupByRequestId(std::vector<std::string> v){
    assert(v.size() == 1);
    name = "groupByRequestId";
    class_name = "Tool-Group-groupByRequestId";
    app_id = 0;
    flow_id = 0;
    cur_request_id = 0;
    final_request_id = std::stoi(v[0])-1;
    first_flag = true;
    Request_Group_Function = true;
    data_type_support.push_back(std::pair<std::vector<uint8_t>, uint8_t>({STRING}, STRING));
}

groupByRequestId::groupByRequestId(short cnt){
    name = "groupByRequestId";
    class_name = "Tool-Group-groupByRequestId";
    final_request_id = cnt - 1;
    app_id = 0;
    flow_id = 0;
    cur_request_id = 0;
    first_flag = true;
    Request_Group_Function = true;
    data_type_support.push_back(std::pair<std::vector<uint8_t>, uint8_t>({STRING}, STRING));
}

groupByRequestId::groupByRequestId(){
    name = "groupByRequestId";
    class_name = "Tool-Group-groupByRequestId";
    app_id = 0;
    flow_id = 0;
    cur_request_id = 0;
    first_flag = true;
    Request_Group_Function = true;
    data_type_support.push_back(std::pair<std::vector<uint8_t>, uint8_t>({STRING}, STRING));
}

groupByRequestId::~groupByRequestId(){

}

void groupByRequestId::registe(){
    Function::Register("Tool-Group-groupByRequestId", createObject);
    Function::Register("Tool-Group-groupByRequestId", createObject2);
}

Function* groupByRequestId::createObject(std::vector<std::string> params){
    return new groupByRequestId(params);
}

Function* groupByRequestId::createObject2(){
    return new groupByRequestId();
}


void groupByRequestId::start(){

}

bool groupByRequestId::waitForResource(){
    return true;
}

//DONE: 需要修改为groupByRequestId
void groupByRequestId::process(std::vector<std::vector<Data*>>& data_input, std::vector<Data*>& data_output){
    // 验证输入个数
    assert(batch_size != 0);
    assert(data_input.size() <= batch_size);
    assert(data_input[0].size() == 1);
    // 处理
    for(int i = 0; i < data_input.size(); i++){
        if(data_input[i][0] == DATA_END){
            assert(false);
        }
        // 如果此id与存储id相同，则加入，不同则赋值
        else if(data_input[i][0]->getRequestId() == cur_request_id){
            if(first_flag){
                app_id = data_input[i][0]->getAppId();
                flow_id = data_input[i][0]->getFlowId();
                first_flag = false;
            }
            if(data_input[i][0]->type == HEAD){  // 是HEAD类型把里面的内容拆出来分组，并释放头结点，否则直接参与分组。
                Data* item = data_input[i][0]->next;
                while(item){
                    cur_datas.push_back(item);
                    item = item->next;
                }
                // 释放头节点内存
                executor->free((char*)data_input[i][0]);
            }else if(data_input[i][0]->type == EMPTY){
                // 释放Empty内存
                executor->free((char*)data_input[i][0]);
            }
        }else{ // 如果不同
            if(first_flag){
                app_id = data_input[i][0]->getAppId();
                flow_id = data_input[i][0]->getFlowId();
                first_flag = false;
            }
            // 将cur_datas中的数据使用链表链起来，放入data_output
            char* ptr = executor->malloc(Data_Memory);
            Data* head = new (ptr)Data(app_id, flow_id, cur_request_id, HEAD);
            Data* p = head;
            for(Data* i : cur_datas){
                p->next = i;
                p = p->next;
            }
            p->next = nullptr;
            data_output.push_back(head);
            short next_id = data_input[i][0]->getRequestId();
            // 补全从cur_request_id，到next_id的空表头
            for(int i = cur_request_id+1; i < next_id; i++){
                char* ptr = executor->malloc(Data_Memory);
                Data* head = new (ptr)Data(app_id, flow_id, i, HEAD);
                data_output.push_back(head);
            }
            cur_request_id = next_id;
            cur_datas.clear();
            // 加当前Data加入cur_datas
            if(data_input[i][0]->type == HEAD){
                Data* item = data_input[i][0]->next;
                while(item){
                    cur_datas.push_back(item);
                    item = item->next;
                }
                // 释放头节点内存
                executor->free((char*)data_input[i][0]);
            }else{
                // 释放Empty内存
                executor->free((char*)data_input[i][0]);
            }
        }
    }
}

bool groupByRequestId::releaseResource(){
    return true;
}

void groupByRequestId::finish(){
    std::vector<Data*> data_output;
    // 将cur_datas中的数据使用链表链起来，放入data_output
    char* ptr = executor->malloc(Data_Memory);
    Data* head = new (ptr)Data(app_id, flow_id, cur_request_id, HEAD);
    Data* p = head;
    for(Data* i : cur_datas){
        p->next = i;
        p = p->next;
    }
    p->next = nullptr;
    data_output.push_back(head);
    // 将一直到final_request_id的空头发出去
    for(int i = cur_request_id+1; i <= final_request_id; i++){
        char* ptr = executor->malloc(Data_Memory);
        Data* head = new (ptr)Data(app_id, flow_id, i, HEAD);
        data_output.push_back(head);
    }
    // 向nextQs发送出去
    sendToQueues(data_output);
}

void groupByRequestId::copy(Function* other){
    Function::defaultCopy(other);
    groupByRequestId* oth = dynamic_cast<groupByRequestId*>(other);
    this->final_request_id = oth->final_request_id;
}

};